#include<db/btree.h>

namespace bpt
{
    BTreeNode *BTree::addNode(const bool enableDeadNode) const
    {
        // ReSharper disable once CppDFAUnreachableCode
        if (buffer_ == nullptr || getSize() >= buffer_size_ / sizeof(BTreeNode)) {
            return nullptr; // No space to add a new node
        }
        // ReSharper disable once CppDFAConstantConditions
        if (enableDeadNode)
        {
            // ReSharper disable once CppDFAUnreachableCode
            BTreeNodeIndexType index = 1;
            while (index < getSize() && !isDeadNode(getNode(index)))
                ++index;
            if (index < getSize())
            {
                return getNode(index);
            }
        }
        setSize(getSize() + 1);
        BTreeNode *newNode = getSize()==1?getRoot():getNode(getSize() - 1);
        if (newNode == nullptr) {
            return nullptr; // Failed to get a new node
        }
        clearNode(newNode);
        return newNode;
    }

    BTree::BTreeResultSet& BTree::BTreeResultSet::operator++()
    {
        if (btree_ == nullptr || btree_->getNode(node_) == nullptr)
        {
            key = -1;
            index_ = 0;
            data = nullptr;
            btree_ = nullptr;
            return *this;
        }
        // 闭区间查询
        if (key == to_)
        {
            node_ = -1;
            key = -1;
            index_ = 0;
            data = nullptr;
            btree_ = nullptr;
            return *this;
        }
        ++index_;
        while (btree_->getNode(node_) != nullptr && index_ >= BTree::getNodeSize(btree_->getNode(node_)))
        {
            node_ = (node_ ? btree_->getNode(node_) : btree_->getRoot())->children[BTree::getNodeSize(btree_->getNode(node_))].node;
            index_ = 0;
        }
        if (btree_->getNode(node_) == nullptr)
        {
            key = -1;
            index_ = 0;
            data = nullptr;
            return *this;
        }
        key = btree_->getNode(node_)->keys[index_];
        data = btree_->getNode(node_)->children[index_].data;
        return *this;
    }

    BTree::BTreeResultSet& BTree::BTreeResultSet::operator++(int)
    {
        BTree::BTreeResultSet& result = *this;
        ++(*this);
        return result;
    }

    BTree::BTreeResultSet& BTree::BTreeResultSet::operator+=(int n)
    {
        if (btree_ == nullptr || btree_->getNode(node_) != nullptr)
            return *this;
        while (--n) // 循环n-1次
        {
            if ((node_ ? btree_->getNode(node_) : btree_->getRoot()) == nullptr || (node_ ? btree_->getNode(node_) : btree_->getRoot())->keys[index_] == to_)
            {
                node_ = -1;
                index_ = 0;
                key = -1;
                data = nullptr;
                return *this;
            }
            ++(*this);
        }

        return ++(*this);
    }


    BTreeNodeSizeType BTree::searchNode(const BTreeNode *parent, const BTreeNode *child) const
    {
        if (parent == nullptr || child == nullptr)
            return -1;
        BTreeNodeSizeType key = 0;
        while (key < getNodeSize(parent) && getNode(parent->children[key].node) != child) {
            ++key;
        }
        return key;
    }

    BTreeNodeSizeType BTree::searchNode(const BTreeNode *parent, const BTreeKeyType key)
    {
        if (parent == nullptr)
            return -1;
        return static_cast<BTreeNodeSizeType>(std::upper_bound(&(parent->keys[0]), &(parent->keys[getNodeSize(parent)]), key)-parent->keys);
    }

    // 找到最小的记录
    // path栈底一定为根节点
    // ReSharper disable once CppDFAConstantFunctionResult
    void * BTree::search(const BTreeKeyType key, std::stack<BTreeNode*> *path) const
    {
        if (getSize() == 0) {
            return nullptr;
        }
        BTreeNode *root = getRoot();
        if (root == nullptr) {
            return nullptr;
        }
        // ReSharper disable once CppDFAUnreachableCode
        if (path!=nullptr)
            path->push(root);
        while (root != nullptr && !isLeafNode(root)) {
            root = getNode(root->children[searchNode(root, key)].node);

            if (path != nullptr)
            {
                path->push(root);
            }
        }
        if (root == nullptr) {
            return nullptr;
        }
        for (BTreeNodeSizeType i = 0; i < getNodeSize(root); ++i) {
            if (root->keys[i] == key) {
                return root->children[i].data;
            }
        }
        return nullptr;
    }

    BTree::BTreeResultSet BTree::search(const BTreeKeyType key1, const BTreeKeyType key2) const
    {
        std::stack<BTreeNode*> path;
        search(key1, &path);
        if (path.empty()) {
            return {};
        }
        BTreeNodeSizeType index = 0;
        while (index < getNodeSize(path.top()) && path.top()->keys[index] < key1) {
            ++index;
        }
        return {this, getNodeIndex(path.top()), index, key2};
    }

    // 中间节点的insertNodeKey必须由子节点split时调用
    std::pair<BTreeKeyType, BTreeNode *> BTree::insertNode(BTreeNode *node, const BTreeKeyType key, void *data) const {
        if (node == nullptr || getNodeSize(node) > BTreeMaxKeys) {
            return std::make_pair(UnusedKey, nullptr); // Node is null or full
        }
        std::pair<BTreeKeyType, BTreeNode *> result = {UnusedKey, nullptr};
        // 如果添加元素后节点大小超过最大值，先分裂再插入(避免插入后再分裂导致每个BTreeNode都要多一位空的children)
        if (getNodeSize(node) == BTreeMaxKeys)
        {
            result = split(node);
            if (result.second == nullptr)
            {
                return result;
            }
            if (key >= result.first)
            {
                node = result.second;
            }
        }
        const BTreeNodeSizeType i = searchNode(node, key);
        node->children[getNodeSize(node) + 1].data = node->children[getNodeSize(node)].data; // Move last child data to the new position
        for (BTreeNodeSizeType j = getNodeSize(node); j > i; --j) {
            node->keys[j] = node->keys[j - 1];
            node->children[j].data = node->children[j - 1].data;
        }
        // 正常情况下插入在i的位置
        node->keys[i] = key;
        node->children[i].data = data;
        if (!isLeafNode(node))
            std::swap(node->children[i], node->children[i+1]);
        setNodeSize(node, getNodeSize(node) + 1);
        return result; // Return the index where the key was inserted
    }

    // ReSharper disable once CppMemberFunctionMayBeConst
    void BTree::insert(const BTreeKeyType key, void *data) { // NOLINT(*-make-member-function-const)
        if (!getSize()) {
            BTreeNode* root = addNode();
            if (root == nullptr) {
                return;
            }
            setLeafNode(root, true);
            setNodeSize(root, 1);
            root->keys[0] = key;
            root->children[0].data = data;
            return;
        }
        std::stack<BTreeNode*> path;
        // TODO: 支持重复Key
        if (nullptr != search(key, &path))
            return; // Key已存在，直接返回，不允许重复Key
        BTreeNode *cursor = path.empty() ? nullptr : path.top();
        path.pop();
        auto result = insertNode(cursor, key, data);
        BTreeNode *parent = path.empty() ? nullptr : path.top();
        if (!path.empty())
            path.pop();
        while (parent!=nullptr && result.second != nullptr) {
            result = insertNode(parent, result.first, (void*)(getNodeIndex(result.second)));
            cursor = parent;
            if (!path.empty()) {
                parent = path.top();
                path.pop();
            } else {
                parent = nullptr;   // cursor为根节点
            }
        }
        if (parent == nullptr && result.second != nullptr) {
            assert (cursor == getRoot());
            parent = addNode();
            memcpy(parent, cursor, sizeof (BTreeNode));
            cursor = parent;
            parent = getRoot();
            clearNode(parent);
            parent->keys[0] = result.first;
            parent->children[0].node = getNodeIndex(cursor);
            parent->children[1].node = getNodeIndex(result.second);
            setNodeSize(parent, 1);
        }

    }

    // 返回需要插入父节点的key和新分裂的节点
    std::pair<BTreeKeyType, BTreeNode *> BTree::split(BTreeNode *cursor) const {
        if (cursor == nullptr)
        {
            return std::make_pair(UnusedKey, nullptr); // Invalid nodes
        }
        BTreeNode *LLeaf = cursor;
        BTreeNode *RLeaf = addNode();
        if (RLeaf == nullptr) {
            return std::make_pair(UnusedKey, nullptr); // Failed to add a new node
        }
        constexpr BTreeNodeSizeType keyIndex = (BTreeMaxKeys + 1) / 2;
        // 取右孩子第一个key作为新的key
        const BTreeKeyType insertKey = cursor->keys[keyIndex];
        const bool withoutKey = !isLeafNode(cursor);

        // 总共只会有BTreeMaxKeys个key
        setNodeSize(RLeaf, getNodeSize(cursor) - keyIndex - withoutKey);
        setLeafNode(RLeaf, isLeafNode(cursor));
        setNodeSize(LLeaf, keyIndex);


        for (BTreeNodeSizeType i = 0; i < getNodeSize(RLeaf); i++) {
            RLeaf->keys[i] = cursor->keys[i + keyIndex + withoutKey];
            RLeaf->children[i] = cursor->children[i + keyIndex + withoutKey];
        }
        if (isLeafNode(cursor)) {
            RLeaf->children[getNodeSize(RLeaf)] = cursor->children[getNodeSize(RLeaf) + keyIndex + withoutKey];
            LLeaf->children[getNodeSize(LLeaf)].node = getNodeIndex(RLeaf);
        } else
        {
            RLeaf->children[getNodeSize(RLeaf)] = cursor->children[getNodeSize(RLeaf) + getNodeSize(LLeaf) + withoutKey];
        }

        for (BTreeNodeSizeType i = getNodeSize(LLeaf); i < BTreeMaxKeys; i++) {
            LLeaf->keys[i] = 0;
            LLeaf->children[i+1].node = -1;
        }
        for (BTreeNodeSizeType i = getNodeSize(RLeaf); i < BTreeMaxKeys; i++)
        {
            RLeaf->keys[i] = 0;
            RLeaf->children[i+1].node = -1;
        }

        return std::make_pair(insertKey, RLeaf);
    }

    bool BTree::combineNode(BTreeNode* parent, BTreeNode* child) const
    {
        if (parent == nullptr || child == nullptr)
            return false;
        BTreeNodeSizeType childIndex = searchNode(parent, child);
        if (childIndex == -1 || childIndex > getNodeSize(parent))
            return false;
        if (getNodeSize(child) > BTreeMaxKeys / 2)
            return false;
        // 启发式合并
        const BTreeNodeSizeType leftSize = childIndex ? getNodeSize(getNode(parent->children[childIndex-1].node)) : 0;
        const BTreeNodeSizeType rightSize = childIndex < getNodeSize(parent) ? getNodeSize(getNode(parent->children[childIndex+1].node)) : 0;
        if (leftSize == rightSize && !leftSize)   // 合并右节点
        {
            return true;    // 无左右孩子，合并父节点
        }
        BTreeNode *rightChild;
        // 向兄弟抢元素
        if (leftSize > BTreeMaxKeys / 2 || rightSize > BTreeMaxKeys / 2)
        {
            if (leftSize > rightSize)
            {
                BTreeNode* leftChild = getNode(parent->children[childIndex - 1].node);
                if (leftChild == nullptr)
                    return false;
                if (!isLeafNode(child))
                {
                    child->children[getNodeSize(child)+1] = child->children[getNodeSize(child)];
                    for (BTreeNodeSizeType i=getNodeSize(child); i > 0; --i)
                    {
                        child->keys[i] = child->keys[i-1];
                        child->children[i].data = child->children[i-1].data;
                    }
                    child->keys[0] = parent->keys[childIndex-1];
                    parent->keys[childIndex-1] = leftChild->keys[getNodeSize(leftChild)-1];
                    child->children[0] = leftChild->children[getNodeSize(leftChild)];
                    setNodeSize(child, getNodeSize(child) + 1);
                    setNodeSize(leftChild, getNodeSize(leftChild) - 1);
                    leftChild->keys[getNodeSize(leftChild)] = 0;
                    leftChild->children[getNodeSize(leftChild)+1].data = nullptr;
                }
                else
                {
                    child->children[getNodeSize(child)+1] = child->children[getNodeSize(child)];
                    for (BTreeNodeSizeType i=getNodeSize(child); i > 0; --i)
                    {
                        child->keys[i] = child->keys[i-1];
                        child->children[i] = child->children[i-1];
                    }
                    child->keys[0] = leftChild->keys[getNodeSize(leftChild)-1];
                    child->children[0] = leftChild->children[getNodeSize(leftChild)];
                    parent->keys[childIndex-1] = child->keys[0];
                    setNodeSize(child, getNodeSize(child) + 1);
                    setNodeSize(leftChild, getNodeSize(leftChild) - 1);
                    leftChild->keys[getNodeSize(leftChild)] = 0;
                    leftChild->children[getNodeSize(leftChild)+1].data = nullptr;
                }
            } else
            {
                rightChild = getNode(parent->children[childIndex+1].node);
                if (rightChild == nullptr)
                    return false;
                // 在child最后面插入rightChild的第一个key
                if (!isLeafNode(child))
                {
                    child->keys[getNodeSize(child)] = parent->keys[childIndex];
                    parent->keys[childIndex] = rightChild->keys[0];
                    child->children[getNodeSize(child)+1].data = rightChild->children[0].data;
                } else
                {
                    child->children[getNodeSize(child)+1] = child->children[getNodeSize(child)];
                    child->keys[getNodeSize(child)] = rightChild->keys[0];
                    child->children[getNodeSize(child)].data = rightChild->children[0].data;
                    parent->keys[childIndex] = rightChild->keys[1];
                }
                setNodeSize(child, getNodeSize(child) + 1);
                for (BTreeNodeSizeType i=0;i<getNodeSize(rightChild)-1;++i)
                    rightChild->keys[i] = rightChild->keys[i+1];
                for (BTreeNodeSizeType i=0;i<getNodeSize(rightChild);++i)
                    rightChild->children[i] = rightChild->children[i+1];
                setNodeSize(rightChild, getNodeSize(rightChild) - 1);
                for (BTreeNodeSizeType i=getNodeSize(rightChild); i < BTreeMaxKeys; ++i)
                {
                    rightChild->keys[i] = 0;
                    rightChild->children[i+1].data = nullptr;
                }
            }
            return false; // 兄弟节点有足够的元素，不需要修改父节点
        }
        if (leftSize > rightSize)   // 和节点数多的一侧合并
        {
            rightChild = child;
            child = getNode(parent->children[childIndex-1].node);
            if (child == nullptr)
                return false;
            --childIndex; // 兄弟节点在左边，childIndex减1，相当于leftChild(child)和child(rightChild)合并
        } else
        {
            rightChild = getNode(parent->children[childIndex+1].node);
            if (rightChild == nullptr)
                return false;
        }
        // 不能借位，合并右节点
        if (isLeafNode(child))
        {
            for (BTreeNodeSizeType i=0;i<getNodeSize(rightChild);++i)
            {
                child->keys[getNodeSize(child)+i] = rightChild->keys[i];
                child->children[getNodeSize(child)+i] = rightChild->children[i];
            }
            setNodeSize(child, getNodeSize(child) + getNodeSize(rightChild));
            // 设置叶子节点的下一个指针
            child->children[getNodeSize(child)] = rightChild->children[getNodeSize(rightChild)];
        } else
        {
            child->keys[getNodeSize(child)] = parent->keys[childIndex];
            for (BTreeNodeSizeType i=0;i<getNodeSize(rightChild);++i)
            {
                child->keys[getNodeSize(child)+i+1] = rightChild->keys[i];
                child->children[getNodeSize(child)+i+1] = rightChild->children[i];
            }
            setNodeSize(child, getNodeSize(child) + getNodeSize(rightChild) + 1);
            child->children[getNodeSize(child)] = rightChild->children[getNodeSize(rightChild)];
        }
        setNodeTombstone(rightChild);
        return removeNode(parent, parent->keys[childIndex]);
    }

    bool BTree::removeNode(BTreeNode *node, const BTreeKeyType key)
    {
        if (node == nullptr || getNodeSize(node) == 0)
            return false;
        const BTreeNodeSizeType childIndex = searchNode(node, key) - 1;
        if (childIndex == -1 || childIndex > getNodeSize(node))
            return false;
        for (BTreeNodeSizeType i=childIndex; i < getNodeSize(node); ++i) {
            node->keys[i] = node->keys[i + 1];
            node->children[i + !isLeafNode(node)] = node->children[i + !isLeafNode(node) + 1];
        }
        setNodeSize(node, getNodeSize(node) - 1);
        for (BTreeNodeSizeType i=getNodeSize(node); i < BTreeMaxKeys; ++i)
        {
            node->keys[i] = 0;
            node->children[i+1].data = nullptr;
        }
        return getNodeSize(node) < BTreeMaxKeys / 2; // 返回父节点是否需要合并
    }

    void BTree::remove(const BTreeKeyType key) const
    {
        if (!getSize())
            return;
        std::stack<BTreeNode*> path;
        if (nullptr == search(key, &path) || path.empty())
            return;
        BTreeNode *cursor = path.top();
        path.pop();
        BTreeNode *parent = path.empty() ? nullptr : path.top();
        if (!path.empty())
            path.pop();
        bool needCombine = removeNode(cursor, key);
        while (parent != nullptr && needCombine)
        {
            needCombine = combineNode(parent, cursor);
            cursor = parent;
            if (!path.empty())
            {
                parent = path.top();
                path.pop();
            }
            else
                parent = nullptr;   // cursor为根节点
        }
        if (parent == nullptr && needCombine)
        {
            assert(cursor == getRoot());
            if (isLeafNode(cursor))
            {
                // 整棵树只有一个根节点了，如果如果还有元素就忽略needCombine信号，否则将根节点删除
                if (!getNodeSize(cursor))
                {
                    clearNode(cursor);
                    setSize(0);
                }
            } else
            {
                // 只有根节点没元素了才更新根节点，不考虑B树的内部节点元素个数约束
                if (!getNodeSize(cursor))
                {
                    cursor = getNode(cursor->children[0].node);
                    memcpy(getRoot(), cursor, sizeof(BTreeNode));
                    setNodeTombstone(cursor);
                }
            }
        }
    }
}

namespace db {

// 上层需要delete iov中的空间
std::pair<struct iovec, unsigned int> BTree::splitNode(DataBlock &cursor)
{
    DataBlock &lLeaf = cursor;
    unsigned int newBlock = addBlock();
    if (newBlock == 0) {
        return {{nullptr, 0}, 0}; // 分裂失败
    }
    DataBlock rLeaf(table_);
    BufDesc *rBuf = kBuffer.borrow(table_->name_.c_str(), newBlock);
    rLeaf.attach(rBuf->buffer);
    constexpr unsigned short BlockHalf = (BLOCK_SIZE - sizeof(DataHeader) - sizeof(Trailer)) / 2; // 一半的大小
    unsigned short count = cursor.getSlots();
    size_t half = 0;
    Slot *slots = cursor.getSlotsPointer();
    unsigned short i;
    for (i = 0; i < count; ++i) {
        half += be16toh(slots[i].length);
        if (half > BlockHalf) {
            break;
        }
    }
    while (lLeaf.getSlots() > i) {
        Record record;
        lLeaf.refSlots(i, record);
        rLeaf.copyRecord(record);
        lLeaf.deallocate(i);
    }
    Record record;
    lLeaf.refSlots(i, record);
    unsigned int keyIndex = table_->info_->key;
    struct iovec iov;
    unsigned char *iovPtr;
    unsigned int iovLen;
    record.refByIndex(&iovPtr, &iovLen, keyIndex);
    iov.iov_base = new unsigned char[iovLen];
    iovPtr = static_cast<unsigned char *>(iov.iov_base);
    record.getByIndex( reinterpret_cast<char *>(iovPtr), &iovLen, keyIndex);
    iov.iov_len = iovLen;
    rLeaf.setNext(lLeaf.getNext());
    lLeaf.setNext(rLeaf.getSelf());
    kBuffer.releaseBuf(rBuf);
    return {iov, newBlock}; // 返回分裂位置和新块ID
}

std::pair<struct iovec, unsigned int> BTree::splitNode(BTreeBlock &cursor)
{
    unsigned short splitPosition = cursor.splitPosition();
    if (splitPosition == -1) {
        return {{nullptr, 0}, 0}; // 分裂位置无效
    }
    BTreeBlock lLeaf = cursor;
    lLeaf.setTable(table_);
    Slot keySlot = lLeaf.getKeySlot(splitPosition);
    struct iovec keyIov = {nullptr, keySlot.length};
    void *keyPtr = new unsigned char[keySlot.length];
    unsigned int keyLen = keySlot.length;
    lLeaf.getKey(splitPosition, keyPtr, &keyLen);
    keyIov.iov_base = keyPtr;
    unsigned int newBlock = addBlock();
    if (newBlock == 0) {
        return {{nullptr, 0}, 0}; // 分裂失败
    }
    BTreeBlock rLeaf(table_);
    BufDesc *rBuf = kBuffer.borrow(table_->name_.c_str(), newBlock);
    rLeaf.attach(rBuf->buffer);
    rLeaf.getTrailerPointer()->data = lLeaf.getChild(splitPosition+1);
    while (lLeaf.getSize() > splitPosition + 1) {
        //struct iovec iov;
        void *iovPtr;
        unsigned int iovLen;
        lLeaf.refKey(splitPosition, &iovPtr, &iovLen);
        rLeaf.appendKey(iovPtr, iovLen, newBlock);
        lLeaf.deallocate(splitPosition);
    }
    return {keyIov, newBlock}; // 返回分裂位置和新块ID
}

std::pair<struct iovec, unsigned int> BTree::insertNode(DataBlock &cursor, const std::vector<struct iovec> &iov)
{
    std::pair<bool, unsigned short> ret = cursor.insertRecord(iov);
    if (ret.first) {
        BufDesc *buf = kBuffer.borrow(table_->name_.c_str(), 0);
        SuperBlock super;
        super.attach(buf->buffer);
        super.setRecords(super.getRecords() + 1);
        super.detach();
        kBuffer.releaseBuf(buf);
        return {{nullptr}, 0}; // 插入成功
    }
    if (ret.second == -1) {
        return {{nullptr}, 0}; // key已经存在
    }

    // 分裂block
    unsigned short insert_position = ret.second;
    std::pair<unsigned short, bool> split_position =
        cursor.splitPosition(Record::size(iov), insert_position);

    // 先分配一个block
    DataBlock next(table_);
    next.setTable(table_);
    unsigned int blkId = table_->allocate();
    BufDesc *bd2 = kBuffer.borrow(table_->name_.c_str(), blkId);
    next.attach(bd2->buffer);

    // 移动记录到新的block上
    Record record;
    while (cursor.getSlots() > split_position.first) {
        cursor.refSlots(split_position.first, record);
        // TODO: 表达式结果未使用
        next.copyRecord(record);
        cursor.deallocate(split_position.first);
    }
    // 插入新记录，不需要再重排顺序
    if (!split_position.second)
        // TODO: 表达式结果未使用
            cursor.insertRecord(iov);
    else
        // TODO: 表达式结果未使用
            next.insertRecord(iov);
    // 维持数据链
    next.setNext(cursor.getNext());
    cursor.setNext(next.getSelf());
    bd2->relRef();

    BufDesc *bd = kBuffer.borrow(table_->name_.c_str(), 0);
    SuperBlock super;
    super.attach(bd->buffer);
    super.setRecords(super.getRecords() + 1);
    bd->relRef();

    struct iovec keyIov;
    unsigned char *keyPtr;
    unsigned int keyLen;
    RelationInfo *info = table_->info_;
    unsigned short keyIndex = info->key;
    bd = kBuffer.borrow(table_->name_.c_str(), blkId);
    next.attach(bd->buffer);
    next.refSlots(0, record);
    record.refByIndex(&keyPtr, &keyLen, keyIndex);
    keyIov.iov_len = keyLen;
    keyIov.iov_base = keyPtr = new unsigned char[keyLen];
    record.getByIndex(reinterpret_cast<char *>(keyPtr), &keyLen, keyIndex);
    record.detach();
    next.detach();
    kBuffer.releaseBuf(bd);

    return {keyIov, blkId}; // 返回是否插入成功和新块ID
}

std::pair<struct iovec, unsigned int> BTree::insertNode(BTreeBlock &cursor, void *buf, size_t len, unsigned int child)
{
    RelationInfo *info = table_->info_;
    DataType *keyType = info->fields[info->key].type;
    unsigned short i = 0;
    BTreeNode *node = cursor.getNodesPointer();
    // struct iovec iov;
    // unsigned char *iovPtr;
    // unsigned int iovLen;
    // cursor.refKey(0, (void **) & iovPtr, &iovLen);
    // while (i < cursor.getSize() && !keyType->less(iovPtr, iovLen, static_cast<unsigned char *>(buf), static_cast<unsigned int>(len))) {
    //     ++i;
    //     cursor.refKey(i, (void **) &iovPtr, &iovLen);
    // }
    i = cursor.searchNode(buf, len);
    unsigned char *ptr = cursor.allocate(static_cast<unsigned short>(len), i, child);
    if (ptr != nullptr) {
        memcpy(ptr, buf, len);
        return {{nullptr}, 0};
    }
    // 分裂block
    std::pair<struct iovec, unsigned int> splitResult = splitNode(cursor);
    if (splitResult.second == 0)
        return {{nullptr}, 0}; // 分裂失败
    if (keyType->less(static_cast<unsigned char *>(buf), static_cast<unsigned int>(len), static_cast<unsigned char *>(splitResult.first.iov_base), static_cast<unsigned int>(splitResult.first.iov_len))) {
        // 插入到左边
        ptr = cursor.allocate(static_cast<unsigned short>(len), i, child);
        if (ptr != nullptr) {
            memcpy(ptr, buf, len);
            return splitResult;
        }
    } else {
        // 插入到右边
        BTreeBlock rightBlock(table_);
        BufDesc *bufDesc = kBuffer.borrow(table_->name_.c_str(), splitResult.second);
        rightBlock.attach(bufDesc->buffer);
        unsigned int rightIndex = rightBlock.searchNode(buf, len);
        ptr = rightBlock.allocate(static_cast<unsigned short>(len), rightIndex, child);
        if (ptr != nullptr) {
            memcpy(ptr, buf, len);
            rightBlock.detach();
            kBuffer.releaseBuf(bufDesc);
            return splitResult; // 插入成功
        }
        rightBlock.detach();
        kBuffer.releaseBuf(bufDesc);
    }
    return {{nullptr}, 0}; // 插入失败
}

bool BTree::search(void *keyBuf, size_t len, std::vector<struct iovec> *iov, std::stack<unsigned short> *path) const
{
    unsigned char rootIndex = getRoot();
    BTreeBlock root(table_);
    BufDesc *buf = kBuffer.borrow(table_->name_.c_str(), rootIndex);
    root.attach(buf->buffer);
    RelationInfo *info = table_->info_;
    unsigned int keyIndex = info->key;
    DataType *keyType = info->fields[keyIndex].type;
    if (path!=nullptr) {
        path->push(rootIndex);
    }
    while (rootIndex != 0 && !root.isLeafNode()) {
        root.detach();
        kBuffer.releaseBuf(buf);
        buf = kBuffer.borrow(table_->name_.c_str(), rootIndex);
        root.attach(buf->buffer);
        if (root.isLeafNode()) break;
        unsigned short childIndex = root.searchNode(keyBuf, len);
        rootIndex = root.getChild(childIndex);
        if (rootIndex == 0) {
            break;
        }
        if (path != nullptr) {
            path->push(rootIndex);
        }
    }
    root.detach();
    if (rootIndex == 0)
        return false;
    //buf = kBuffer.borrow(table_->name_.c_str(), rootIndex);
    DataBlock data(table_);
    Record record;
    data.attach(buf->buffer);
    if (!data.refRecord( record, keyBuf, static_cast<unsigned int>(len)).first) {
        data.detach();
        kBuffer.releaseBuf(buf);
        return false;
    }
    if (iov == nullptr) {
        data.detach();
        kBuffer.releaseBuf(buf);
        return true; // 只需要检查key是否存在
    }
    unsigned char header[Record::HEADER_SIZE];
    record.ref(*iov, header);
    if (!record.get(*iov, header)) {
        data.detach();
        kBuffer.releaseBuf(buf);
        return false;
    }
    data.detach();
    kBuffer.releaseBuf(buf);
    return true;
}
void BTree::insert(const std::vector<struct iovec> &iov)
{
    unsigned short rootIndex = getRoot();
    BTreeBlock block(table_);
    BufDesc *buf;
    // TODO: 插入第一个元素
    std::stack<unsigned short> path;

    auto info = table_->info_;
    unsigned int keyIndex = info->key;

    size_t len = iov[keyIndex].iov_len;
    void * keyBuf = iov[keyIndex].iov_base;

    if (search(keyBuf, len, nullptr, &path))
        return;     // key已存在
    unsigned short cursorIndex = path.top();
    path.pop();
    unsigned short parentIndex = path.empty() ? 0 : path.top();
    if (!path.empty())
        path.pop();
    // insert
    buf = kBuffer.borrow(table_->name_.c_str(), cursorIndex);
    DataBlock data(table_);
    data.attach(buf->buffer);
    std::pair<struct iovec, unsigned int> result = insertNode(data, iov);
    block.detach();
    kBuffer.releaseBuf(buf);

    // buf = kBuffer.borrow(table_->name_.c_str(), 0);
    SuperBlock super;
    // super.attach(buf->buffer);
    // super.setRecords(super.getRecords() + 1);
    // kBuffer.releaseBuf(buf);
    if (result.second == 0) {
        delete[] static_cast<unsigned char *>(result.first.iov_base);
        return; // 插入成功
    }
    while (parentIndex != 0 && result.second != 0) {
        buf = kBuffer.borrow(table_->name_.c_str(), parentIndex);
        block.attach(buf->buffer);
        auto newResult = insertNode(block, result.first.iov_base, result.first.iov_len, result.second);
        delete[] static_cast<unsigned char *>(result.first.iov_base);
        result = newResult;
        block.detach();
        kBuffer.releaseBuf(buf);

        if (!path.empty()) {
            parentIndex = path.top();
            path.pop();
        } else {
            parentIndex = 0; // 到达根节点
        }
    }
    if (parentIndex == 0 && result.second != 0) {
        parentIndex = addBlock();
        buf = kBuffer.borrow(table_->name_.c_str(), parentIndex);
        block.attach(buf->buffer);
        block.getTrailerPointer()->data = htobe32(rootIndex); // 设置根节点的子节点
        block.appendKey(result.first.iov_base, result.first.iov_len, result.second);
        //BTreeNode *nodes = block.getNodesPointer();
        //nodes[0].data = cursorIndex;
        //nodes[1].data = result.second;
        block.detach();
        kBuffer.releaseBuf(buf);
        buf = kBuffer.borrow(table_->name_.c_str(), 0);
        super.attach(buf->buffer);
        super.setFirstIndex(parentIndex);
        super.detach();
        kBuffer.releaseBuf(buf);
    }
    delete[] static_cast<unsigned char *>(result.first.iov_base);
}

void BTree::update(const std::vector<struct iovec> &iov)
{
    RelationInfo *info = table_->info_;
    unsigned int keyIndex = info->key;
    DataType *keyType = info->fields[keyIndex].type;
    remove(iov[keyIndex].iov_base, iov[keyIndex].iov_len); // 删除旧记录
    insert(iov); // 插入新记录
}

bool BTree::removeNode(DataBlock &cursor, void *keyBuf, size_t len)
{
    unsigned short index = cursor.searchRecord(keyBuf, len);
    if (index == -1) {
        return false; // key不存在
    }
    cursor.deallocate(index);
    return cursor.getFreeSize() < (BLOCK_SIZE - sizeof(DataHeader) - sizeof(Trailer)) / 2; // 返回是否需要合并
}

bool BTree::removeNode(BTreeBlock &cursor, void *keyBuf, size_t len)
{
    unsigned short index = cursor.searchNode(keyBuf, len);
    if (index == -1) {
        return false; // key不存在
    }
    cursor.deallocate(index);
    return cursor.getFreeSize() < (BLOCK_SIZE - sizeof(BTreeHeader) - sizeof(BTreeTrailer)) / 2; // 返回是否需要合并
}

bool BTree::combineNode(BTreeBlock &parent, DataBlock &child, unsigned short keyIndex)
{
    static constexpr unsigned short FullBlockSize = BLOCK_SIZE - sizeof(DataHeader) - sizeof(Trailer);
    static constexpr unsigned short HalfBlockSize = FullBlockSize / 2;

    unsigned short childIndex = parent.searchNode(child.getSelf());
    if (childIndex == -1 || childIndex > parent.getSize())
        return false; // 无效的childIndex
    if (child.getFreeSize() > (BLOCK_SIZE - sizeof(DataHeader) - sizeof(Trailer)) / 2)
        return false; // child节点不需要合并
    // 启发式合并
    BufDesc *buf;
    DataBlock block(table_);
    unsigned short leftBlockIndex = 0, rightBlockIndex = 0;
    unsigned short leftSize = -1, rightSize = -1;
    if (childIndex) {
        leftBlockIndex = parent.getChild(childIndex - 1);
        buf = kBuffer.borrow(table_->name_.c_str(), leftBlockIndex);
        block.attach(buf->buffer);
        leftSize = FullBlockSize - block.getFreeSize();
        block.detach();
        kBuffer.releaseBuf(buf);
    }
    if (childIndex < parent.getSize()) {
        rightBlockIndex = parent.getChild(childIndex + 1);
        buf = kBuffer.borrow(table_->name_.c_str(), rightBlockIndex);
        block.attach(buf->buffer);
        rightSize = FullBlockSize - block.getFreeSize();
        block.detach();
        kBuffer.releaseBuf(buf);
    }
    if (!leftBlockIndex && !rightBlockIndex) {
        return true; // 没有左右兄弟节点，直接合并父节点
    }
    // 向兄弟抢元素
    const bool leftHasSpace = leftSize > HalfBlockSize && leftSize + (FullBlockSize - child.getFreeSize()) > HalfBlockSize;
    const bool rightHasSpace = rightSize > HalfBlockSize && rightSize + (FullBlockSize - child.getFreeSize()) > HalfBlockSize;
    if (leftHasSpace || rightHasSpace) {
        if (leftSize > rightSize) {
            // 从左兄弟节点借位
            buf = kBuffer.borrow(table_->name_.c_str(), leftBlockIndex);
            block.attach(buf->buffer);
            while (child.getFreeSize() > HalfBlockSize) {
                Record record;
                block.refSlots(block.getSlots()-1, record);
                child.copyRecord(record);
                block.deallocate(block.getSlots() - 1);
            }
            //struct iovec keyIov;
            unsigned char *keyPtr;
            unsigned int keyLen;
            Record record;
            child.refSlots(0, record);
            record.refByIndex(&keyPtr, &keyLen, keyIndex);
            parent.updateKey(childIndex-1, keyPtr, keyLen);
            block.detach();
            kBuffer.releaseBuf(buf);
        } else {
            // 从右兄弟节点借位
            buf = kBuffer.borrow(table_->name_.c_str(), rightBlockIndex);
            block.attach(buf->buffer);
            while (child.getFreeSize() > HalfBlockSize) {
                Record record;
                block.refSlots(0, record);
                child.copyRecord(record);
                block.deallocate(0);
            }
            block.detach();
            kBuffer.releaseBuf(buf);
        }
        return false; // 不需要合并父节点
    }
    // 合并兄弟节点
    if (leftSize > rightSize) { // 和左兄弟节点合并
        buf = kBuffer.borrow(table_->name_.c_str(), leftBlockIndex);
        block.attach(buf->buffer);
        while (child.getSlots() > 0) {
            Record record;
            child.refSlots(0, record);
            block.copyRecord(record);
            child.deallocate(0);
        }
        block.setNext(child.getNext());
        block.detach();
        kBuffer.releaseBuf(buf);
        // 删除child block
        table_->deallocate(child.getSelf());
        // struct iovec keyIov;
        void *keyPtr;
        unsigned int keyLen;
        // 删除父节点的key(leftChild位置)
        parent.refKey(childIndex-1, &keyPtr, &keyLen);
        return removeNode(parent, keyPtr, keyLen); // 删除父节点的key
    } else { // 和右兄弟节点合并
        buf = kBuffer.borrow(table_->name_.c_str(), rightBlockIndex);
        block.attach(buf->buffer);
        while (child.getFreeSize() > HalfBlockSize) {
            Record record;
            block.refSlots(0, record);
            child.copyRecord(record);
            block.deallocate(0);
        }
        // 设置child的下一个指针
        child.setNext(block.getNext());
        block.detach();
        kBuffer.releaseBuf(buf);
        // 删除rightChild block
        table_->deallocate(rightBlockIndex);
        // struct iovec keyIov;
        void *keyPtr;
        unsigned int keyLen;
        parent.refKey(childIndex, &keyPtr, &keyLen);
        return removeNode(parent, keyPtr, keyLen); // 删除父节点的key
    }
}
bool BTree::combineNode(BTreeBlock &parent, BTreeBlock &child)
{
    static constexpr unsigned short FullBlockSize = BLOCK_SIZE - sizeof(DataHeader) - sizeof(Trailer);
    static constexpr unsigned short HalfBlockSize = FullBlockSize / 2;

    unsigned short childIndex = parent.searchNode(child.getSelf());
    if (childIndex == -1 || childIndex > parent.getSize())
        return false; // 无效的childIndex
    if (child.getFreeSize() > (BLOCK_SIZE - sizeof(DataHeader) - sizeof(Trailer)) / 2)
        return false; // child节点不需要合并
    // 启发式合并
    BufDesc *buf;
    BTreeBlock block(table_);
    unsigned short leftBlockIndex = 0, rightBlockIndex = 0;
    unsigned short leftSize = -1, rightSize = -1;
    if (childIndex) {
        leftBlockIndex = parent.getChild(childIndex - 1);
        buf = kBuffer.borrow(table_->name_.c_str(), leftBlockIndex);
        block.attach(buf->buffer);
        leftSize = FullBlockSize - block.getFreeSize();
        block.detach();
        kBuffer.releaseBuf(buf);
    }
    if (childIndex < parent.getSize()) {
        rightBlockIndex = parent.getChild(childIndex + 1);
        buf = kBuffer.borrow(table_->name_.c_str(), rightBlockIndex);
        block.attach(buf->buffer);
        rightSize = FullBlockSize - block.getFreeSize();
        block.detach();
        kBuffer.releaseBuf(buf);
    }
    if (!leftBlockIndex && !rightBlockIndex) {
        return true; // 没有左右兄弟节点，直接合并父节点
    }
    // 向兄弟抢元素
    const bool leftHasSpace = leftSize > HalfBlockSize && leftSize + (FullBlockSize - child.getFreeSize()) > HalfBlockSize;
    const bool rightHasSpace = rightSize > HalfBlockSize && rightSize + (FullBlockSize - child.getFreeSize()) > HalfBlockSize;
    if (leftHasSpace || rightHasSpace) {
        if (leftSize > rightSize) {
            // 从左兄弟节点借位
            buf = kBuffer.borrow(table_->name_.c_str(), leftBlockIndex);
            block.attach(buf->buffer);
            // struct iovec keyIov;
            void *keyPtr;
            unsigned int keyLen;
            parent.refKey(childIndex - 1, &keyPtr, &keyLen);
            while (child.getFreeSize() > HalfBlockSize) {
                child.appendKey(keyPtr, keyLen, block.getTrailerPointer()->data);
                block.refKey(block.getSize()-1, &keyPtr, &keyLen);
                block.deallocate(block.getSize() - 1);
            }
            parent.updateKey(childIndex-1, keyPtr, keyLen);
            block.detach();
            kBuffer.releaseBuf(buf);
        } else {
            // 从右兄弟节点借位
            buf = kBuffer.borrow(table_->name_.c_str(), rightBlockIndex);
            block.attach(buf->buffer);
            // struct iovec keyIov;
            void *keyPtr;
            unsigned int keyLen;
            parent.refKey(childIndex, &keyPtr, &keyLen);
            while (child.getFreeSize() > HalfBlockSize) {
                auto result = insertNode(child, keyPtr, keyLen, block.getNodesPointer()->data);
                if (result.first.iov_base != nullptr) {
                    delete[] static_cast<unsigned char *>(result.first.iov_base);
                }
                block.refKey(0, &keyPtr, &keyLen);
                // 手动更新block的size和freeSize
                block.setFreeSize(block.getFreeSize() + be16toh(block.getNodesPointer()->key.length));
                block.setFreeSpace(block.getFreeSpace() - be16toh(block.getNodesPointer()->key.length));
                block.setSize(block.getSize() - 1);
            }
            parent.updateKey(childIndex, keyPtr, keyLen);
            block.detach();
            kBuffer.releaseBuf(buf);
        }
        return false; // 不需要合并父节点
    }
    // 合并兄弟节点
    if (leftSize > rightSize) { // 和左兄弟节点合并
        buf = kBuffer.borrow(table_->name_.c_str(), leftBlockIndex);
        block.attach(buf->buffer);
        // struct iovec keyIov;
        void *keyPtr;
        unsigned int keyLen;
        parent.refKey(childIndex-1, &keyPtr, &keyLen);
        unsigned char *ptr = block.allocate(static_cast<unsigned short>(keyLen), block.getSize(), child.getChild(0));
        if (ptr != nullptr) {
            memcpy(ptr, keyPtr, keyLen);
        }
        while (child.getSize() > 0) {
            child.refKey(0, &keyPtr, &keyLen);
            ptr = block.allocate(static_cast<unsigned short>(keyLen), block.getSize(), child.getChild(0));
            if (ptr != nullptr) {
                memcpy(ptr, keyPtr, keyLen);
            }
        }
        block.detach();
        kBuffer.releaseBuf(buf);
        // 删除child block
        table_->deallocate(child.getSelf());
        // 删除父节点的key(leftChild位置)
        parent.refKey(childIndex-1, &keyPtr, &keyLen);
        return removeNode(parent, keyPtr, keyLen); // 删除父节点的key
    } else { // 和右兄弟节点合并
        buf = kBuffer.borrow(table_->name_.c_str(), rightBlockIndex);
        block.attach(buf->buffer);
        // struct iovec keyIov;
        void *keyPtr;
        unsigned int keyLen;
        parent.refKey(childIndex, &keyPtr, &keyLen);
        unsigned char *ptr = child.allocate(static_cast<unsigned short>(keyLen), child.getSize(), block.getTrailerPointer()->data);
        if (ptr != nullptr) {
            memcpy(ptr, keyPtr, keyLen);
        }
        while (block.getSize() > 0) {
            block.refKey(0, &keyPtr, &keyLen);
            ptr = child.allocate(static_cast<unsigned short>(keyLen), child.getSize(), block.getChild(0));
            if (ptr != nullptr) {
                memcpy(ptr, keyPtr, keyLen);
            }
        }
        block.detach();
        kBuffer.releaseBuf(buf);
        // 删除rightChild block
        table_->deallocate(rightBlockIndex);
        parent.refKey(childIndex, &keyPtr, &keyLen);
        return removeNode(parent, keyPtr, keyLen); // 删除父节点的key
    }
}

void BTree::remove(void *keyBuf, size_t len) {
    if (table_ == nullptr)
        return;
    std::stack<unsigned short> path;
    if (!search(keyBuf, len, nullptr, &path) || path.empty())
        return; // key不存在
    unsigned int cursorIndex = path.top();
    path.pop();
    unsigned int parentIndex = path.empty() ? 0 : path.top();
    if (!path.empty())
        path.pop();

    BufDesc *buf1 = kBuffer.borrow(table_->name_.c_str(), cursorIndex), *buf2;
    DataBlock data(table_);
    data.attach(buf1->buffer);
    bool needCombine = removeNode(data, keyBuf, len);
    data.detach();
    kBuffer.releaseBuf(buf1);

    SuperBlock super;
    buf1 = kBuffer.borrow(table_->name_.c_str(), 0);
    super.attach(buf1->buffer);
    super.setRecords(super.getRecords() - 1);
    super.detach();
    kBuffer.releaseBuf(buf1);

    if (!needCombine) {
        return; // 删除成功，不需要合并父节点
    }
    bool isDataBlock = true;

    BTreeBlock parent(table_), child(table_);
    while (parentIndex != 0 && needCombine) {
        buf1 = kBuffer.borrow(table_->name_.c_str(), cursorIndex);
        buf2 = kBuffer.borrow(table_->name_.c_str(), parentIndex);
        parent.attach(buf2->buffer);
        if (isDataBlock) {
            isDataBlock = false;
            data.attach(buf1->buffer);
            needCombine = combineNode(parent, data, table_->info_->key);
        }else {
            child.attach(buf1->buffer);
            needCombine = combineNode(parent, child);
        }
        cursorIndex = parentIndex;
        if (!path.empty()) {
            parentIndex = path.top();
            path.pop();
        } else {
            parentIndex = 0; // 到达根节点
        }
    }
    if (parentIndex == 0 && needCombine) {
        Block block;
        buf1 = kBuffer.borrow(table_->name_.c_str(), cursorIndex);
        block.attach(buf1->buffer);
        if (block.getType() == BLOCK_TYPE_DATA) {
            block.detach();
            data.attach(buf1->buffer);
            if (data.getSlots() == 0) {
                // 整棵树只有一个根节点了，如果如果还有元素就忽略needCombine信号，否则将根节点删除
                buf2 = kBuffer.borrow(table_->name_.c_str(), 0);
                super.attach(buf2->buffer);
                super.setFirstIndex(cursorIndex);
                super.detach();
                kBuffer.releaseBuf(buf2);
            }
            data.detach();
            kBuffer.releaseBuf(buf1);
        } else {
            block.detach();
            child.attach(buf1->buffer);
            if (child.getSize() == 0) {
                buf2 = kBuffer.borrow(table_->name_.c_str(), 0);
                super.attach(buf2->buffer);
                super.setFirstIndex(child.getTrailerPointer()->data);
            }
        }
    }
}
}
